package com.example.service;

import com.example.latch.MySqlASyncLatch;
import com.example.latch.MysqlLatchListener;
import com.example.po.SysServer;
import com.example.utils.TimeUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import javax.sql.DataSource;
import java.util.Date;
import java.util.List;
import java.util.concurrent.*;

@Service
public class SysClusterServerServiceImpl implements SysClusterServerService, MysqlLatchListener, InitializingBean, DisposableBean {

    Logger logger = LoggerFactory.getLogger(getClass());

    @Resource
    private SysServerService sysServerService;

    @Resource
    private SysMQMessageService sysMQMessageService;

    @Resource
    private DataSource dataSource;

    //每次抓取总数
    @Value("${mq.message.max-size:10}")
    private int maxSize;

    //心跳频率 默认秒
    @Value("${server.refresh.period:10}")
    private int period;

    //心跳超时的心跳间隔次数
    @Value("${server.refresh.period.size:3}")
    private int periodSize;

    //心跳线程池的核心数
    @Value("${server.refresh.core:10}")
    private int corePoolSize;

    //心跳单位 默认单位:秒
    private TimeUnit timeUnit = TimeUnit.SECONDS;

    private MySqlASyncLatch mySqlASyncLatch;

    private ExecutorService executorService;

    private ScheduledThreadPoolExecutor scheduledThreadPoolExecutor;

    //运行状态
    private volatile boolean running = true;

    //是否主节点
    private volatile boolean isMaster = false;

    //启动的时间戳
    private volatile Date startTime = null;

    /****
     * 启动本节点的注册
     * @throws Exception
     */
    @Override
    public void afterPropertiesSet() throws Exception {
        this.startTime = new Date();
        this.executorService = Executors.newCachedThreadPool();
        this.scheduledThreadPoolExecutor = new ScheduledThreadPoolExecutor(corePoolSize);
        this.scheduledThreadPoolExecutor.scheduleAtFixedRate(new ServerRefreshThread(), 0, period, timeUnit);
        this.mySqlASyncLatch = new MySqlASyncLatch(this.dataSource, this);
        this.mySqlASyncLatch.start();
    }

    /***
     * 关闭线程回收资源
     * @throws Exception
     */
    @Override
    public void destroy() throws Exception {
        this.running = false;
        this.mySqlASyncLatch.stop();
        this.executorService.shutdownNow();
        this.scheduledThreadPoolExecutor.shutdownNow();
    }

    /****
     * 主节点执行,大于3个心跳间隔就可以认为是节点掉线了。
     */
    @Override
    public void isLeader() {
        logger.warn("server: " + sysServerService.getCurrentSysServer() + " 被选举为主节点 ...");
        this.setMaster(true);
        while (running) {
            for (SysServer sysServer : this.sysServerService.getAllServer()) {
                if (!TimeUtils.isLated(sysServer.getRefreshTime(), Math.negateExact(period * periodSize))) {
                    sysServerService.removeById(sysServer.getId());
                    if (sysServer != null && (sysServer.getId() != (sysServerService.getCurrentServerKey()))) {
                        int totalSize = sysMQMessageService.getSysMQMessageTotalSize(sysServer.getId());
                        if ((maxSize * 3) <= totalSize) {
                            List<SysServer> allServers = sysServerService.getAllServer();
                            for (int i = 0; i < allServers.size(); i++) {
                                Long activeServerKey = allServers.get(i).getId();
                                sysMQMessageService.updateMessageServerBatch(sysServer.getId(), activeServerKey, i, allServers.size());
                            }
                        } else {
                            Long activeServerKey = sysServerService.getRandomServerKey();
                            sysMQMessageService.updateMessageServer(sysServer.getId(), activeServerKey);
                        }
                        logger.warn("server节点: " + sysServer.getId() + " 已经掉线 ...");
                    }
                }
            }
            sysMQMessageService.updateMessageServer(sysServerService.getRandomServerKey(), (maxSize * 2));
        }
    }

    @Override
    public void setMaster(boolean master) {
        this.isMaster = master;
    }

    //节点刷新线程
    class ServerRefreshThread implements Runnable {

        @Override
        public void run() {
            SysServer sysServer = new SysServer();
            sysServer.setId(sysServerService.getCurrentServerKey());
            sysServer.setHost(sysServerService.getHost());
            sysServer.setPort(sysServerService.getPort());
            sysServer.setMaster(isMaster);
            sysServer.setStartTime(startTime);
            sysServer.recalculateFields();
            sysServerService.saveOrUpdate(sysServer);
        }
    }
}
